package kafka

import (
	"github.com/IBM/sarama"
	"github.com/sirupsen/logrus"
	"os"
)

type Message struct {
	Data  string
	Topic string
}

var (
	client  sarama.SyncProducer // 全局的kafka producer对象
	log     *logrus.Logger
	msgChan chan *sarama.ProducerMessage
)

func init() {
	log = logrus.New()
	// 设置日志输出为os.Stdout
	log.Out = os.Stdout
	log.Level = logrus.DebugLevel
	// 可以设置像文件等任意`io.Writer`类型作为日志输出
	// file, err := os.OpenFile("logrus.log", os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
	// if err == nil {
	//  log.Out = file
	// } else {
	//  log.Info("Failed to log to file, using default stderr")
	// }

	log.Info("kafka:init log success")
}

// Init 是初始化kafka的函数
func Init(addrs []string, chanSize int) (err error) {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Successes = true

	client, err = sarama.NewSyncProducer(addrs, config)
	if err != nil {
		log.Errorf("producer closed,err:", err)
	}

	// 初始化MsgChan
	msgChan = make(chan *sarama.ProducerMessage, chanSize)
	//  起一个后台的goroutineMsgChan中读数据
	go sendMsg()

	return
}

func sendMsg() {
	/**
	for msg := range MsgChan {
		kafkaMsg := &sarama.ProducerMessage{}
		kafkaMsg.Topic = msg.Topic
		kafkaMsg.Value = msg.Value
		pid, offset, err := client.SendMessage(kafkaMsg)
		if err != nil {
			log.Warnf("send msg failed,err%v\n", err)
			continue
		}
		log.Infof("send msg success,pid:%v offset:%v\n", pid, offset)
	}
	*/
	for {
		select {
		case msg := <-msgChan:
			pid, offset, err := client.SendMessage(msg)
			if err != nil {
				logrus.Warning("send msg failed,err:", err)
				return
			}
			logrus.Infof("send msg to kafka success. pid:%v offset:%v", pid, offset)
		}
	}
}

// 把msgChan私有，暴露一个方法让外部能访问
func ToMsgChan(msg *sarama.ProducerMessage) {
	msgChan <- msg
}
